package 链接snappydate数据库相关操作.snappydateDDL相关操作

import org.apache.spark.sql.SnappySession
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.DriverManager
/**
 * 很奇怪，链接snappydate只需地址正确就行，不需要输入啥账户密码
 * 链接重点：val url: String = s"jdbc:snappydata://172.16.136.26:端口"
 *      val conn = DriverManager.getConnection(url)
 *      之后就可以对snappydate进行DDL操作
 * connect(strURL:String)函数是链接snappydate集群
 * select_snappydate()函数是查询snappydate数据
 *insert_snappydate()是插入数据库数据操作
 *create_drop_table()是删除创建表
 * */
class 链接Snappydate数据库及相关操作 {
  val url: String = s"jdbc:snappydata://172.16.136.26:端口"
  val conn = DriverManager.getConnection(url)
  def connect(strURL:String): Unit = {
    /**此为集群环境中优雅链接snappydate
     * 达到不抛异常的效果
     * 使用时只需将strURL:String换成172.16.136.26:端口 就行
     * */
    try {
      val conf: SparkConf = new SparkConf()
        .setAppName("APP")
        .set("spark.ui.port", "8080")
        .set("spark.snappydata.connection", strURL)
        .setMaster("local[*]")
        .set("spark.sql.warehouse.dir", "/tmp/warehouse")
      snappySession = new SnappySession(new SparkContext(conf))
    } catch {
      case ex: Exception =>
        ex.printStackTrace()
    }
  }
  def select_snappydate(): Unit ={
    /**查询snappydate示例
     * */
    val sql_c = s"select * from app.audit_log where (select Count(1) from app.audit_struct_table where audit_struct_table.objectname= audit_log.objectname) = 0"
    val pre_c = conn.prepareStatement(sql_c)
    val rst_c = pre_c.executeQuery()
    while(rst_c.next())
    {
      println(rst_c.getString("schemaname"),rst_c.getString("objectname"),
        rst_c.getString("objecttype"),rst_c.getString("lastupdatedate"),
        rst_c.getString("objectcreatescript"))

    }
    pre_c.close()
    rst_c.close()
  }
  def insert_snappydate(): Unit ={
    /**
     * 这是第一套查询代码，先通过数据表查询，然后再把相关数据插入进去
     * 重点：直接获取数据表中的列名然后插入*/
    val sql_c = s"select * from app.audit_log where (select Count(1) from app.audit_struct_table where audit_struct_table.objectname= audit_log.objectname) = 0"
    val pre_c = conn.prepareStatement(sql_c)
    val rst_c = pre_c.executeQuery()
    val stmt_c = conn.prepareStatement("INSERT INTO APP.AUDIT_LOG(schemaname,objecttype,objectname,lastupdatedate," +
      "objectcreatescript,status) VALUES('" + rst_c.getString("schemaname") + "','" +
      rst_c.getString("objecttype") + "','" + rst_c.getString("objectname") +
      "','" + rst_c.getString("lastupdatedate") +
      "','" + rst_c.getString("objectcreatescript") + "','DELETE')")
    stmt_c.execute()
    stmt_c.close()

    /**
     * 插入数据表的第二套代码：
     * 像普通数据库一样操作就行
     * */
      val tableSchema,tableType,tableName,millis,clob=null;

    val stmt = conn.prepareStatement("INSERT INTO APP.AUDIT_LOG(schemaname,objecttype,objectname,lastupdatedate," +
      "objectcreatescript,status) VALUES('" + tableSchema + "','" + tableType + "','" + tableName +
      "','" + millis +
      "','" + clob + "','init')")
    stmt.execute()
    stmt.close()
  }
  def create_drop_table(): Unit = {
    /**
     * 此为创建表和删除表*/
    val stmt1 = conn.createStatement()
    // Creating a table (PARTSUPP) using JDBC connection
    //drop table
    stmt1.execute("DROP TABLE IF EXISTS app.audit_struct_table")
    println("Database drop successfully")
    //create table
    stmt1.execute("CREATE TABLE app.audit_struct_table ( " +
      "schemaname     VARCHAR NOT NULL," +
      "objectname     VARCHAR NOT NULL"+")")

  }

}
